package com.devilvan.customOutputFormat;

import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.nio.charset.StandardCharsets;

public class CustomOutputFormat extends FileOutputFormat<Text, NullWritable> {
    @Override
    public RecordWriter<Text, NullWritable> getRecordWriter(TaskAttemptContext context) throws IOException, InterruptedException {
        Path goodComment = new Path("D:\\Mine\\开课吧大数据-Hadoop\\2、MR\\1、数据&代码资料\\9、自定义outputFormat\\数据\\result1011\\goodCommon.txt");
        Path badComment = new Path("D:\\Mine\\开课吧大数据-Hadoop\\2、MR\\1、数据&代码资料\\9、自定义outputFormat\\数据\\result1011\\badCommon.txt");
        FileSystem fileSystem = FileSystem.get(context.getConfiguration());
        FSDataOutputStream goodStream = fileSystem.create(goodComment);
        FSDataOutputStream badStream = fileSystem.create(badComment);
        return new OutputRecordWriter(goodStream, badStream);
    }

    static class OutputRecordWriter extends RecordWriter<Text,NullWritable> {
        FSDataOutputStream goodStream = null;
        FSDataOutputStream badStream = null;

        public OutputRecordWriter(FSDataOutputStream goodStream, FSDataOutputStream badStream) {
            this.goodStream = goodStream;
            this.badStream = badStream;
        }

        @Override
        public void write(Text key, NullWritable value) throws IOException, InterruptedException {
            String[] fields = key.toString().split("\t");
            if ("0".equals(fields[9])){
                goodStream.write((key + "\r\n").getBytes(StandardCharsets.UTF_8));
            } else if ("1".equals(fields[9])){
                badStream.write((key + "\r\n").getBytes(StandardCharsets.UTF_8));
            }
        }

        @Override
        public void close(TaskAttemptContext context) throws IOException, InterruptedException {
            IOUtils.closeStreams(goodStream,badStream);
        }
    }
}
